package com.atguigu.flink.chapter09;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.List;
import java.util.Map;


public class Flink02_CEP_BaseUse_Loop {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

           SingleOutputStreamOperator<WaterSensor> waterStream = env.readTextFile("input/sensor.txt")
                   .map(new MapFunction<String, WaterSensor>() {
                       @Override
                       public WaterSensor map(String value) throws Exception {
                           String[] data = value.split(",");
                           return new WaterSensor(
                                   data[0],
                                   Long.valueOf(data[1]),
                                   Integer.valueOf(data[2])
                           );
                       }
                   }).assignTimestampsAndWatermarks(
                        WatermarkStrategy
                            .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                            .withTimestampAssigner(((element, recordTimestamp) -> element.getTs()))
                   );


           //2.定义模式
           Pattern<WaterSensor, WaterSensor> pattern = Pattern
                   //模式名字只能是字符，不能包含符号，且独一无二
                   .<WaterSensor>begin("start")
                   .where(new SimpleCondition<WaterSensor>() {
                       @Override
                       public boolean filter(WaterSensor value) throws Exception {
                           return "sensor_1".equals(value.getId());
                       }
                   })
                  // .times(2);  // 严格两次
                //    .times(2,4);   //2，3，4次
                   //  .oneOrMore() 一次或多次

            .timesOrMore(2)  //两次或2次以上 ，比上一行，更多的结果，对于无界流，必须加一个终止条件until()

           .until(new SimpleCondition<WaterSensor>() {
               @Override
               public boolean filter(WaterSensor value) throws Exception {
                   return "sensor_2".equals(value.getId());
               }
           });


           //3.把模式作用在流上，得到一个模式流
           PatternStream<WaterSensor> ps = CEP.pattern(waterStream, pattern);


           //4.从模式流中获取匹配到的数据
           ps.select(new PatternSelectFunction<WaterSensor, String>() {
               @Override
               public String select(Map<String, List<WaterSensor>> map) throws Exception {

                   //一条数据匹配一次，返回一个值，list是为了那些一次性匹配多个的，预留的情况
                   return map.toString();
               }
           })
                   .print();

               try {
                   env.execute();
               } catch (Exception e) {
                   e.printStackTrace();
               }


           }
}
